package serv_impl

import (
	"bufio"
	"context"
	"time"

	"atomgit.com/openlinksaas/proto-gen-go.git/k8s_proxy_api"
	"github.com/google/uuid"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/client-go/rest"
)

type LogItem struct {
	LogId        string
	Cancel       context.CancelFunc
	LogChan      chan []byte
	LastReadTime int64
}

func (apiImpl *K8SProxyApiImpl) OpenLog(ctx context.Context, req *k8s_proxy_api.OpenLogRequest) (*k8s_proxy_api.OpenLogResponse, error) {
	tokenInfo, err := getTokenInfo(apiImpl.linkSaasAddr, req.Token)
	if err != nil {
		return nil, err
	}
	k8sClientSet, err := getK8sClientSet(apiImpl.kubeConfigFile)
	if err != nil {
		return nil, err
	}
	{
		if req.Namespace == "" {
			return &k8s_proxy_api.OpenLogResponse{
				Code:   k8s_proxy_api.OpenLogResponse_CODE_NO_PERMISSION,
				ErrMsg: "必须指定命名空间",
			}, nil
		}
		client := k8sClientSet.CoreV1().Namespaces()
		res, err := client.Get(ctx, req.Namespace, metav1.GetOptions{})
		if err != nil {
			return nil, err
		}
		if res.Labels[PROJECT_LABEL] != tokenInfo.ProjectId {
			return &k8s_proxy_api.OpenLogResponse{
				Code:   k8s_proxy_api.OpenLogResponse_CODE_NO_PERMISSION,
				ErrMsg: "命名空间没有关联项目",
			}, nil
		}
	}
	if !tokenInfo.Admin {
		userPerm := apiImpl.getResourceUserPerm(ctx, req.Namespace, req.ResourceType, req.ResourceName, tokenInfo.MemberUserId)
		if !userPerm.UpdateScale {
			return &k8s_proxy_api.OpenLogResponse{
				Code:   k8s_proxy_api.OpenLogResponse_CODE_NO_PERMISSION,
				ErrMsg: "没有权限",
			}, nil
		}
	}

	tailLines := int64(100)
	client := k8sClientSet.CoreV1().Pods(req.Namespace)
	request := client.GetLogs(req.PodName, &corev1.PodLogOptions{
		Container:  req.ContainerName,
		Follow:     true,
		Timestamps: true,
		TailLines:  &tailLines,
	})

	logId := uuid.NewString()
	logCtx, cancel := context.WithCancel(context.Background())
	logChan := make(chan []byte, 100)
	apiImpl.logItemMap.Store(logId, &LogItem{
		LogId:        logId,
		Cancel:       cancel,
		LogChan:      logChan,
		LastReadTime: time.Now().Unix(),
	})
	go apiImpl.readLog(logCtx, request, logChan)

	return &k8s_proxy_api.OpenLogResponse{
		Code:  k8s_proxy_api.OpenLogResponse_CODE_OK,
		LogId: logId,
	}, nil
}

func (apiImpl *K8SProxyApiImpl) readLog(ctx context.Context, request *rest.Request, logChan chan<- []byte) {
	defer func() {
		recover()
	}()

	readCloser, err := request.Stream(context.TODO())
	if err != nil {
		return
	}

	defer readCloser.Close()
	r := bufio.NewReader(readCloser)
	for {
		bytes, err := r.ReadBytes('\n')
		if len(bytes) > 0 {
			logChan <- bytes
		}
		if err != nil {
			return
		}
	}
}

func (apiImpl *K8SProxyApiImpl) ReadLog(ctx context.Context, req *k8s_proxy_api.ReadLogRequest) (*k8s_proxy_api.ReadLogResponse, error) {
	defer func() {
		recover()
	}()

	logItemValue, ok := apiImpl.logItemMap.Load(req.LogId)
	if !ok {
		return &k8s_proxy_api.ReadLogResponse{
			Code:   k8s_proxy_api.ReadLogResponse_CODE_NO_LOG,
			ErrMsg: "日志句柄不存在",
		}, nil
	}
	logItem := logItemValue.(*LogItem)
	logItem.LastReadTime = time.Now().Unix()
	apiImpl.logItemMap.Store(logItem.LogId, logItem)

	select {
	case data := <-logItem.LogChan:
		return &k8s_proxy_api.ReadLogResponse{
			Code: k8s_proxy_api.ReadLogResponse_CODE_OK,
			Data: string(data),
		}, nil
	case <-time.After(30 * time.Second):
		return &k8s_proxy_api.ReadLogResponse{
			Code: k8s_proxy_api.ReadLogResponse_CODE_OK,
			Data: "",
		}, nil
	}
}
